import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 * 来监控传感器温度值，将温度值低于 32F 的温度输出到 side output
 */
object SideOutputTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val stream: DataStream[String] = env.socketTextStream("localhost", 7777)

    val tempDS = stream.map(
      data => {
        val dataArr = data.split(",")
        TemperatureData(dataArr(0), dataArr(1).toLong, dataArr(2).toDouble)
      }
    )

    //TODO FreezingMonitor
    val monitoredReadings: DataStream[TemperatureData] = tempDS
      .process(new FreezingMonitor)

    monitoredReadings
      .getSideOutput(new OutputTag[String]("freezing-alarms"))
      .print()

    tempDS.print()

  }
  case class TemperatureData(id: String, time: Long, temp: Double)

  class FreezingMonitor extends ProcessFunction[TemperatureData,  TemperatureData] {

    // 定义一个侧输出标签
    lazy val freezingAlarmOutput: OutputTag[String] = new OutputTag[String]("freezing-alarms")

    override def processElement(value: TemperatureData,
                                ctx: ProcessFunction[TemperatureData, TemperatureData]#Context,
                                out: Collector[TemperatureData]): Unit = {
      // 温度在 32F 以下时，输出警告信息
      if (value.temp < 32.0){
        ctx.output(freezingAlarmOutput, s"Freezing Alarm for ${value.id}")
      }
      //所有数据直接常规输出到主流
      out.collect(value)
    }
  }
}
